package com.liuhm.core;


import cn.hutool.core.io.FileUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.liuhm.core.thread.TrackThreadPoolTaskExecutor;
import com.liuhm.entity.TableField;
import com.liuhm.utils.CsvUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 *  生产者消费者模式 抽象类
 * @author liuhaomin
 * @date 2020/10/30
 */
@Slf4j
public abstract class AbstractBroker {

	/**
	 * 创建 读线程的线程池
	 */
	private Integer csvReadThreadNum;

	public void setCsvReadThreadNum(Integer csvReadThreadNum) {
		this.csvReadThreadNum = csvReadThreadNum;
	}

	public Integer getCsvReadThreadNum() {
		return csvReadThreadNum!=null?csvReadThreadNum:5;
	}


	/**
	 * 导入成功记录数
	 */
	private AtomicLong writeCount = new AtomicLong(0L);

	/**
	 * 有界阻塞队列
	 */
	private final LinkedBlockingQueue<List<String>> queue = new LinkedBlockingQueue<>(1000);

	/**
	 * 获取队列
	 * @return
	 */
	public LinkedBlockingQueue<List<String>> getQueue() {
		return queue;
	}

	/**
	 * 生产者-读数据到有界队列
	 * @param file
	 * @param startRow
	 * @param columnSize
	 * @throws Exception
	 */
	public abstract void producer(File file, int startRow, int columnSize) throws Exception;

	/**
	 * 消费者-消费有界队列中的数据
	 */
	public void consumer(File file,List<TableField> tableFieldList){

		List<Integer> floatIndexList = null;
		try {
			// 创建表
			// 获取浮点数（小数）下标索引
			floatIndexList = getFloatIndexByFieldType(tableFieldList);

		} catch (Exception e) {
			log.info("获取表信息异常：{}", e);
		}



		// 多线程写入数据
		int cpuCount = getSystemCpuCount();
		int threadCount = cpuCount > 8 ? 10 : 5;

		// 定义线程池前缀
		ThreadFactory customThreadFactory = new ImportThreadFactory("User:" +   "-" + "import");
		TrackThreadPoolTaskExecutor taskExecutor = new TrackThreadPoolTaskExecutor();
		taskExecutor.setThreadFactory(customThreadFactory);
		taskExecutor.setCorePoolSize(threadCount);
		taskExecutor.initialize();
		// 倒计数器
		CountDownLatch threadExitTag = new CountDownLatch(threadCount);

		// 如果当前队列 queue 是否有数据，没有数据等待读线程获取数据
		if(queue.size() == 0){
			log.info("数据文件较大，写线程等待读线程获取数据60s");
			long startTime = System.currentTimeMillis();
			while (true){
				long endTime = System.currentTimeMillis();
				long wait = endTime - startTime;
				if(wait > 60 * 1000 || queue.size() > 0){
					log.info("队列已有数据/已等待60s，退出循环等待；queueSize={}，wait={}", queue.size(), wait);
					break;
				}
			}
		}
		Long startTime=System.currentTimeMillis();
		for (int i = 0; i < threadCount; i++) {
			List<Integer> finalFloatIndexList = floatIndexList;
			taskExecutor.execute(() -> {
				String threadName = Thread.currentThread().getName();
				log.info("=========子线程:{}========start", threadName);
				AtomicLong threadWriteCount = new AtomicLong(0L);
				List<List<String>> dataList = Lists.newArrayList();
				try {
					while (true) {
						// 导入过程遇到异常是否继续
						// 根据key获取当前导入是否取消

						// 从队列取出，等待5s
						List<String> rowList = queue.poll(5, TimeUnit.SECONDS);
						if(null != rowList){
							// 格式化浮点数（小数）字符串
							formatFloatString(rowList, finalFloatIndexList);
							dataList.add(rowList);
							if (dataList.size() >= 50){
								// 获取到 50 条数据后批量插入
//								log.info("打印了{}条 ",dataList.size());
								// 写入100条记录成功，更新文件写入成功数据
								writeCount.addAndGet(50);
								threadWriteCount.addAndGet(50);
								// 清空 dataList
								dataList.clear();
							}
						}else {
							break;
						}
					}

					// 不足100条数据批量写入
					if (dataList.size() > 0) {
						log.info("子线程{} to end，不足50条 dataList size = {}", threadName, dataList.size());
						// 写入记录成功，更新文件写入成功数据
						writeCount.addAndGet(dataList.size());
						threadWriteCount.addAndGet(dataList.size());
						// 清空 dataList
						dataList.clear();
					}
					log.info("写----------数据导入子线程:{} 写入条数", threadName,threadWriteCount.get());
				}  catch (InterruptedException e) {
					log.error("数据导入子线程{}，数据写入过程中-中断异常:{}", threadName, e);
					// 清理中断状态
					Thread.currentThread().interrupt();
					// 标记导入过程中发生异常
				} catch (Exception e){
					log.error("数据导入子线程{}，数据写入过程中-中断异常:{}", threadName, e);
					// 标记导入过程中发生其他未知异常。避免没有其他后续清理工作：清理缓存，备份表等
				}finally {
					// 该线程写工作结束，编辑线程信号量
					log.info("数据导入子线程:{} to end，信号量减少 1", threadName);
					threadExitTag.countDown();
					log.info("=========线程:{}======end", threadName);
				}
			});
		}

		// 阻塞等待所有子线程执行完毕
		try {
			threadExitTag.await();
		} catch (InterruptedException e) {
			log.error("主线程等待所有子线程执行完成异常：{}", e);
		}
		log.info("数据导入 线程:{} 信号量总数={}", Thread.currentThread().getName(), threadExitTag.getCount());
		log.info("==================子线程全部执行完==================");

		// 清理线程池
		taskExecutor.destroy();
		// 更新数据文件写入数据量
		log.info("更新了条数{}",writeCount.get());
		// 更新写完状态
		Long endTime=System.currentTimeMillis();
		log.info("调试打印信息:\n time={}",endTime-startTime);
		// 删除分割后的文件
		String path=file.getAbsolutePath().substring(0,file.getAbsolutePath().lastIndexOf("."));
//		FileUtil.del(path);

		// 导入结束
	}


	/**
	 * 获取系统cpu个数
	 * @return
	 */
	public int getSystemCpuCount(){
		return Runtime.getRuntime().availableProcessors();
	}

	/**
	 * 根据字段类型获取字段类型为小数的下标
	 * @param tableFieldList
	 * @return
	 */
	private List<Integer> getFloatIndexByFieldType(List<TableField> tableFieldList){
		List<Integer> floatIndexList = Lists.newArrayList();
		for (int i = 0; i < tableFieldList.size(); i++) {
			TableField tableField = tableFieldList.get(i);
			if(tableField.getFieldType().contains("decimal")){
				floatIndexList.add(i);
			}
		}
		return floatIndexList;
	}


	/**
	 * 将浮点数字符串格式化：去掉首位多余的0
	 * @param list
	 * @return
	 */
	private void formatFloatString(List<String> list, List<Integer> floatIndexList){
		String str;
		for (int i = 0; i < list.size(); i++) {
			// 类型为浮点数
			if(floatIndexList.contains(i)){
				str =  CsvUtils.formatFloat(list.get(i));
				list.set(i, str);
			}
		}
	}

	/**
	 * 自定义线程工厂
	 */
	public static class ImportThreadFactory implements ThreadFactory{
		private static final AtomicInteger poolNumber = new AtomicInteger(1);
		private ThreadGroup group;
		private final AtomicInteger threadNumber = new AtomicInteger(1);
		private final String namePrefix;

		public ImportThreadFactory(String namePrefix) {
			SecurityManager s = System.getSecurityManager();
			this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
			this.namePrefix = namePrefix + "-" + poolNumber.getAndIncrement() + "-thread-";
		}

		@Override
		public Thread newThread(Runnable r) {
			Thread t = new Thread( r,namePrefix + threadNumber.getAndIncrement());
			if (t.isDaemon()){
				t.setDaemon(true);
			}
			if (t.getPriority() != Thread.NORM_PRIORITY){
				t.setPriority(Thread.NORM_PRIORITY);
			}
			return t;
		}
	}

}
